package com.alibaba.rocketmq.example.simple;

import com.alibaba.rocketmq.client.consumer.MQPullConsumer;
import com.alibaba.rocketmq.client.consumer.MQPullConsumerScheduleService;
import com.alibaba.rocketmq.client.consumer.PullResult;
import com.alibaba.rocketmq.client.consumer.PullTaskCallback;
import com.alibaba.rocketmq.client.consumer.PullTaskContext;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.message.MessageQueue;
import com.alibaba.rocketmq.common.protocol.heartbeat.MessageModel;

public class PullScheduleService {

	public static void main(String[] args) throws MQClientException {
		// 1、初始化MQPullConsumerScheduleService类，
		final MQPullConsumerScheduleService scheduleService = new MQPullConsumerScheduleService("GroupName1");

		scheduleService.setMessageModel(MessageModel.CLUSTERING);

		// 3、 以topic为key值将自定义的PullTaskCallback 对象存入MQPullConsumerScheduleService. callbackTable:ConcurrentHashMap<String /*topic*/,PullTaskCallback>变量中
		scheduleService.registerPullTaskCallback("TopicTest1", new PullTaskCallback() {

			// 2、自定义PullTaskCallback接口的实现类，
			@Override
			public void doPullTask(MessageQueue mq, PullTaskContext context) {
				MQPullConsumer consumer = context.getPullConsumer();
				try {
					// 获取从哪里拉取
					// 获取MessageQueue队列的消费进度
					long offset = consumer.fetchConsumeOffset(mq, false);
					if (offset < 0)
						offset = 0;

					// 在指定的队列和指定的开始位置读取消息内容；再然后对获取到的消息进行相关的业务逻辑处理
					PullResult pullResult = consumer.pull(mq, "*", offset, 32);
					System.out.println(offset + "\t" + mq + "\t" + pullResult);
					switch (pullResult.getPullStatus()) {
						case FOUND :
							break;
						case NO_MATCHED_MSG :
							break;
						case NO_NEW_MSG :
						case OFFSET_ILLEGAL :
							break;
						default :
							break;
					}

					// 存储Offset，客户端每隔5s会定时刷新到Broker
					// 进行消费进度的更新，其中offset值是在获取消息内容时返回的下一个消费进度值；
					consumer.updateConsumeOffset(mq, pullResult.getNextBeginOffset());

					// 设置再过100ms后重新拉取
					context.setPullNextDelayTimeMillis(100);
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		});

		scheduleService.start();
	}
}
